package com.sailing.lianxi.demoservice.thread;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import com.sailing.lianxi.common.Constants;

/**
 * 多线程消费者
 * @author Administrator
 *
 */
public class ThreadConsumer{

	static String TOPIC =Constants.TOPIC_NAME;
	
	
	/**
	 * 手动偏移控制
	 */
	public  static void manualOffsetControl(){
		Properties props = new Properties();
		//kafka服务器地址  
		props.put("bootstrap.servers", Constants.KAFKA_SERVERS);  
		//kafka组
        props.put("group.id", "wg14");  
        /*
        earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 
        latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 
        none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常 
                 以上配置只对一个组里没有消费的消息起作用
        */
        props.put("auto.offset.reset", "earliest");  
        props.put("enable.auto.commit", "false");  
        //Consumer向集群发送自己的心跳，超时则认为Consumer已经死了，kafka会把它的分区分配给其他进程,默认6秒
        props.put("session.timeout.ms", "6000");//毫秒 
        //key序列化器
        String deserialClass = StringDeserializer.class.getName();
        props.put("key.deserializer", deserialClass);  
        //值序列化器
        props.put("value.deserializer", deserialClass);
        
        //设置最大拉取SIZE
        props.put("fetch.message.max.bytes",1024 *1024*100);//设置最大拉取SIZE
//        props.put("max.poll.records", 1);
	     //最小提交记录数
	     for(int i=0;i<Constants.KAFKA_PARTITION_NUM;i++) {
	    	
	    	 new Thread(new Runnable() {
//	 			@Override
	 			public void run() {
	 				//创建消费者
				     KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
				     //订阅主题列表topic
				     consumer.subscribe(Arrays.asList(TOPIC));
	 				 while(true){
	 			    	 ConsumerRecords<String, String> records = consumer.poll(1000);//1秒获取一次
//	 			    	 System.out.println("count="+records.count());
	 			    	 //必须遍历分区，否则多线程消费时，会出现一个线程消费多个分区现象
	 			    	 for(TopicPartition partition: records.partitions()) {
	 			    		for (ConsumerRecord<String, String> record : records.records(partition)) {
		 			    		//　正常这里应该使用线程池处理，不应该在这里处理
		 			    		 System.out.printf(Thread.currentThread().getName()+",partition=%d,offset=%d,key=%s,value=%s",record.partition(),record.offset(),record.key(),record.value()+"\n");
	 			    		}
	 			    	 }
	 			    	 //手动提交
	 			    	 consumer.commitSync();
	 			     }
	 			}
	 		}).start();
	     }
	}
	
	public static void main(String[] args) {
		manualOffsetControl();
	}
}
